package com.xujianlong.day02;

import com.xujianlong.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import javax.lang.model.element.VariableElement;
import java.util.Random;

public class Flink04_Source_Custom {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> dataStreamSource = env.addSource(new mySource());
        dataStreamSource.print();

        env.execute();
    }

    private static class mySource implements ParallelSourceFunction<WaterSensor> {
        private Boolean isRunning = true;
        private Random random = new Random();

        @Override
        public void run(SourceContext sourceContext) throws Exception {
            while (isRunning) {
                sourceContext.collect(new WaterSensor("sensor" + random.nextInt(100), System.currentTimeMillis(), random.nextInt(100)));
                Thread.sleep(1000);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }



}